package com.hucais.etl.history.service

import java.util

import cn.hutool.core.date.DateUtil
import com.hucais.core.constant.Constants
import com.hucais.core.utils.{CommonUtils, DefaultPropertiesUtil}
import com.hucais.etl.common.bean.{CandidateCategory, DwsBookQueryInfo}
import com.hucais.etl.common.dao.MysqlDao
import com.hucais.etl.common.service.{CommonQueryService, JavaCommonService}
import com.hucais.etl.history.bean.HistoryBcBook
import com.hucais.etl.history.dao.HiveDao
import org.apache.spark.SparkContext
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.{Dataset, SparkSession}
import org.elasticsearch.spark.sql.EsSparkSQL

object HistoryBcService {

  def action(ssc: SparkContext, sparkSession: SparkSession): Unit = {
    // 获取候选分类名单
    val categoryList = MysqlDao.getCandidateCategoryList
    val categoryListBd: Broadcast[util.List[CandidateCategory]] = ssc.broadcast(categoryList)

    // 获取ES数据
    val bookDS: Dataset[HistoryBcBook] = HiveDao.getBcBook(sparkSession)
    bookDS.cache()

    // 组装Result数据输出到ES
    val resultDS = initResultData(sparkSession, bookDS, categoryListBd)
    try {
      EsSparkSQL.saveToEs(resultDS, DefaultPropertiesUtil.get("dws.book.query.info"))
    } catch {
      case e: Exception => e.printStackTrace()
    }
  }

  private def initResultData(sparkSession: SparkSession, bookDS: Dataset[HistoryBcBook], categoryListBd: Broadcast[util.List[CandidateCategory]]): Dataset[DwsBookQueryInfo] = {
    import sparkSession.implicits._
    bookDS.mapPartitions(partitions => {
      partitions
        .filter(item => CommonUtils.isNotBlankExt(item.isbn) && CommonUtils.isNotBlankExt(item.book_name) &&
          CommonUtils.isNotBlankExt(item.author) && CommonUtils.isNotBlankExt(item.publishing_house) &&
          CommonUtils.isNotBlankExt(item.pricing) && CommonUtils.isNotBlankExt(item.sell_price) && CommonUtils.isNotBlankExt(item.publishing_time))
        .map(item => {
          // 基础数据清洗
          val bookComments = CommonUtils.handleBlankNumber(item.comment_number)
          val format = CommonUtils.handleBlankStr(item.formats)
          val sellingPrince = CommonUtils.handleBlankFloat(item.sell_price)
          val edition = CommonUtils.handleBlankStr(item.edition)
          val suits = CommonUtils.handleBlankStr(item.number_of_suit)
          val impression = CommonUtils.handleBlankStr(item.impression)
          val bindingLayout = CommonUtils.handleBlankStr(item.binding_layout)

          // 归类书籍分类
          val categoryTmp = CommonUtils.handleBlankStr(item.category)
          val category = JavaCommonService.categorizeBookData(categoryTmp, categoryListBd.value)

          // 过滤无效字符
          var pages: java.lang.Long = null
          if (CommonUtils.isNotBlankExt(item.number_of_pages)) {
            pages = CommonUtils.handleBlankNumber(item.number_of_pages.replaceAll("页", "").replaceAll(" ", ""))
          }
          val publishingHouse = item.publishing_house.replaceAll("<span class=\"text-value\">", "");
          val storePricing = CommonUtils.handleBlankFloat(CommonUtils.filterSomeStr(item.pricing).replaceAll(" ", ""))
          // 过滤敏感字符
          val bookName = CommonUtils.filterSomeStr(item.book_name)
          /*var description = "未知"
          if (CommonUtils.isNotBlankExt(item.book_description)) {
            description = CommonUtils.filterSomeStr(item.book_description)
          }*/
          val description = "未知"

          var slogan = "未知"
          if (CommonUtils.isNotBlankExt(item.slogan)) {
            slogan = CommonUtils.filterSomeStr(item.slogan)
          }

          //计算评论数和相应的积分
          val commentsTuple = CommonQueryService.calCommentsRangeAndIntegral(bookComments)
          val commentsRange = commentsTuple._1
          val commentsIntegral = commentsTuple._2

          // 计算溢价率和相应的积分
          val premiumTuple = CommonQueryService.calPremiumRangeAndIntegral(BigDecimal(sellingPrince), BigDecimal(storePricing))
          val premiumRange = premiumTuple._1
          val premiumIntegral = premiumTuple._2

          //计算出版年限和相应的积分
          val publishedTuple = CommonQueryService.calPublishedYearRangeAndIntegral(item.publishing_time)
          val publishedYearRange = publishedTuple._1
          val publishedYearIntegral = publishedTuple._2

          //计算在售商家数量和相应的积分
          val sellingStoresRange = Constants.NONMAL_STR_VAL
          val sellingStoresIntegral = Constants.NORMAL_NUMBER_VAL

          val totalIntegral = commentsIntegral + premiumIntegral + publishedYearIntegral + sellingStoresIntegral

          val currentTime = DateUtil.format(new java.util.Date(), "yyyy-MM-dd HH:mm:ssZ").replace(" ", "T")
          DwsBookQueryInfo(
            "逆向选品", "中图网", null, null, item.isbn, bookName, category, slogan, description, null,
            null, bookComments, storePricing, sellingPrince, publishingHouse, item.publishing_time, null, null, edition, impression,
            null, null, item.author, null, format, null, suits, bindingLayout, pages, null,
            0, publishedYearRange, publishedYearIntegral, commentsRange, commentsIntegral, premiumRange, premiumIntegral, sellingStoresRange, sellingStoresIntegral, totalIntegral,
            Constants.CREATE_TYPE_ETL, currentTime, currentTime
          )
        })
    })
  }

}
